package cn.doitedu.kafka.day02

import java.util.Properties

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}
import org.apache.kafka.common.serialization.StringSerializer

/**
 * 生产者的ACK应答机制
 *
 * ACK机制可以保证生产者写入的数据可以安全的写入到Kafka中
 * acks = 1普通模式，Producer会将数据发送给Leader分区，只要Leader分区将数据写入磁盘成功就响应Producer
 * 写入的速度相对比较快，也比较安全，如果失败，可以重新发送，重新发送的最大次数，可以通过reties配置
 *
 * acks = 0 渣男模式，Producer会将数据发送给Leader分区，不管Leader是不是将数据写入到本地磁盘，
 * 发送数据快，但是没法保证数据的安全性
 *
 * acks = -1 / all  暖男模式，Producer会将数据发送给Leader分区，Leader分区将数据写入磁盘成功，并且等待其他Follower分区也将数据写入成功，才会向Producer应答
 * 发送数据比较慢，但是安全，如果超时，没有发送成功，会重新发送，
 *
 *
 */
object ProducerAckDemo {

  def main(args: Array[String]): Unit = {


    //指定生产者的参数
    val properties = new Properties()
    //broker的地址
    // 连接kafka节点
    properties.setProperty("bootstrap.servers", "node-1.51doit.cn:9092,node-2.51doit.cn:9092,node-3.51doit.cn:9092")
    //指定key序列化方式
    properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    //指定value序列化方式
    properties.setProperty("value.serializer", classOf[StringSerializer].getName) // 两种写法都行

    //生产者在不同的场景下（数据安全性，三种场景），可以配置相应的ACK应答机制
    //普通模式，Producer会将数据发送给Leader分区，只要Leader分区将数据写入磁盘成功就响应Producer
    //写入的速度相对比较快，也比较安全，如果失败，可以重新发送，重新发送的最大次数，可以通过reties配置
    properties.setProperty("acks", "1")
    //发送数据失败最大重试次数
    properties.setProperty("reties", "100")

    //渣男模式，Producer会将数据发送给Leader分区，不管Leader是不是将数据写入到本地磁盘，
    //发送数据快，但是没法保证数据的安全性
    //properties.setProperty("acks", "0")

    //暖男模式，Producer会将数据发送给Leader分区，Leader分区将数据写入磁盘成功，并且等待其他Follower分区也将数据写入成功，才会向Producer应答
    //发送数据比较慢，但是安全，如果超时，没有发送成功，会重新发送，
    //properties.setProperty("acks", "-1") //-1 或all重新发送的最大次数，可以通过reties配置

    //topic名称
    val topic = "test"
    //创建Kafka生产者
    val producer = new KafkaProducer[String, String](properties)

    //将数据封装到ProducerRecord中
    for(i <- 6001 to 8000) {

      //生成写入Kafka中的数据，用固定的字符串拼接上i
      //如果创建ProducerRecord，没有指定分区编号，默认的写入Topic分区的策略为轮询
      //但是不是写一条就切换，而是达到指定的大小才切换
      //目的是为了将数据均匀的写入到多个分区中，实现读写的负载均衡
      val record = new ProducerRecord[String, String](topic, "my data : "  + i)
      //发送数据
      producer.send(record)

    }

    producer.close()





  }

}
